package com.lili.collection.mq.runner;

import com.lili.collection.core.config.InputConfig;
import com.lili.collection.core.context.PipeContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.plugins.InputPlugin;
import com.lili.collection.core.runner.result.ErrOutputData;
import com.lili.collection.core.runner.result.RunnerResult;
import com.lili.collection.core.runner.stage.AbstractStageRunner;
import com.lili.collection.core.runner.StageRunner;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * 通用输入Runner
 *
 * @author lili
 * @date 2023/3/7 11:18
 */
@Slf4j
public class AsyncInputRunner extends AbstractStageRunner {

    private InputPlugin<InputConfig> plugin;

    private RunnerResult inputRunnerResult;

    private RunnerMsgTransfer transfer;

    @Override
    public RunnerResult getRunnerResult() {
        return this.inputRunnerResult;
    }

    @Override
    public void setRunnerTransfer(RunnerMsgTransfer transfer) {
        this.transfer=transfer;
    }

    @Override
    public RunnerMsgTransfer getRunnerTransfer() {
        return this.transfer;
    }

    @Getter
    private PipeContext pipeContext;

    public AsyncInputRunner(PipeContext pipeContext) {
        this.pipeContext = pipeContext;
        this.inputRunnerResult=new RunnerResult();
    }

    @Override
    public void run() {
        try {
            this.transfer.addThread(Thread.currentThread());
            ensurePlugin();
            this.plugin.execute(this);
        }catch (Throwable throwable){
            log.error("获取pulsar消息失败!", throwable);
            this.transfer.setRunError(true);
            this.transfer.interruptRunner();
            this.inputRunnerResult.setSuccessed(false);
            ErrOutputData errOutputData = this.inputRunnerResult.getErrOutputData();
            errOutputData.setErrorMessage(throwable.toString());
            errOutputData.setType("PULSAR");
        }finally {
            if(Thread.currentThread().isInterrupted()){
                Thread.interrupted();
            }
            this.dispose();
        }
    }

    @Override
    public StageRunner.StageType getStageType() {
        return StageRunner.StageType.Input;
    }

    @Override
    public void dispose() {
        if (this.plugin != null)
            this.plugin.dispose();
    }

    @Override
    public void pause() {
    }

    @Override
    public void resume() {
    }

    @Override
    public void stop() {
        this.dispose();
    }

    private void ensurePlugin() {
        if (this.plugin != null) return;
        InputConfig inputConfig = this.pipeContext.getContext().getInputConfig();
        this.plugin = inputConfig.getPlugin();
        this.plugin.init(inputConfig);
    }
}
